package net.bw.realtime.warehouse.dwd;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bw.realtime.warehouse.dwd
 * @Author: zhangleqing
 * @CreateTime: 2025-08-25  20:31
 * @Description: TODO
 * @Version: 1.0
 */
public class Ower {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

/*
CREATE TABLE ods_ower(
    id STRING,
    roomId INT,
    owerCode STRING,
    owerName STRING,
    owerSex STRING,
    owerAge INT,
    owerIdcard STRING,
    owerTel STRING,
    owerHomeplace STRING,
    owerWorkplace STRING,
    remark STRING,
    createTime TIMESTAMP(3),
    operatorTime TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED,
    WATERMARK FOR createTime AS createTime - INTERVAL '5' SECOND
) WITH (
    'connector' = 'mysql-cdc', -- 使用 MySQL CDC 连接器
    'hostname' = 'node101', -- MySQL 服务器主机名
    'port' = '3306', -- MySQL 服务端口
    'username' = 'root', -- 数据库用户名
    'password' = '123456', -- 数据库密码
    'database-name' = 'sca', -- 数据库名称
    'table-name' = 'ower', -- 源表名称
    'connect.timeout' = '60s' -- 连接超时时间（60秒）
)
*/
        // 1.读取MySQL表数据
        tEnv.executeSql(
                "CREATE TABLE ods_ower(\n" +
                        "    id STRING,\n" +
                        "    roomId INT,\n" +
                        "    owerCode STRING,\n" +
                        "    owerName STRING,\n" +
                        "    owerSex STRING,\n" +
                        "    owerAge INT,\n" +
                        "    owerIdcard STRING,\n" +
                        "    owerTel STRING,\n" +
                        "    owerHomeplace STRING,\n" +
                        "    owerWorkplace STRING,\n" +
                        "    remark STRING,\n" +
                        "    createTime TIMESTAMP(3),\n" +
                        "    operatorTime TIMESTAMP(3),\n" +
                        "    PRIMARY KEY (id) NOT ENFORCED,\n" +
                        "    WATERMARK FOR createTime AS createTime - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        "    'connector' = 'mysql-cdc', -- 使用 MySQL CDC 连接器\n" +
                        "    'hostname' = 'node101', -- MySQL 服务器主机名\n" +
                        "    'port' = '3306', -- MySQL 服务端口\n" +
                        "    'username' = 'root', -- 数据库用户名\n" +
                        "    'password' = '123456', -- 数据库密码\n" +
                        "    'database-name' = 'sca', -- 数据库名称\n" +
                        "    'table-name' = 'ower', -- 源表名称\n" +
                        "    'connect.timeout' = '60s' -- 连接超时时间（60秒）\n" +
                        ")"
        );

/*
create table ods_ower_clean(
    id STRING,
    roomId INT,
    owerCode STRING,
    owerName STRING,
    owerSex STRING,
    owerAge INT,
    owerIdcard STRING,
    owerTel STRING,
    owerHomeplace STRING,
    owerWorkplace STRING,
    remark STRING,
    createTime TIMESTAMP(3),
    operatorTime TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED,
    WATERMARK FOR createTime AS createTime - INTERVAL '5' SECOND
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'ods_ower_clean',
    'properties.bootstrap.servers' = 'node101:9092',
    'key.format' = 'json',
    'value.format' = 'json',
    'sink.delivery-guarantee' = 'exactly-once',
    'properties.transaction.timeout.ms' = '900000',
    'sink.parallelism' = '3'
)
*/
        // 2.创建Kafka数据源表
        tEnv.executeSql(
                "create table ods_ower_clean(\n" +
                        "    id STRING,\n" +
                        "    roomId INT,\n" +
                        "    owerCode STRING,\n" +
                        "    owerName STRING,\n" +
                        "    owerSex STRING,\n" +
                        "    owerAge INT,\n" +
                        "    owerIdcard STRING,\n" +
                        "    owerTel STRING,\n" +
                        "    owerHomeplace STRING,\n" +
                        "    owerWorkplace STRING,\n" +
                        "    remark STRING,\n" +
                        "    createTime TIMESTAMP(3),\n" +
                        "    operatorTime TIMESTAMP(3),\n" +
                        "    PRIMARY KEY (id) NOT ENFORCED,\n" +
                        "    WATERMARK FOR createTime AS createTime - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        "    'connector' = 'upsert-kafka',\n" +
                        "    'topic' = 'ods_ower_clean',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'key.format' = 'json',\n" +
                        "    'value.format' = 'json',\n" +
                        "    'sink.delivery-guarantee' = 'exactly-once',\n" +
                        "    'properties.transaction.timeout.ms' = '900000',\n" +
                        "    'sink.parallelism' = '3'\n" +
                        ")"
        );

/*
insert into ods_ower_clean
select
    id,
    roomId,
    owerCode,
    owerName,
    owerSex,
    owerAge,
    owerIdcard,
    owerTel,
    owerHomeplace,
    owerWorkplace,
    remark,
    createTime,
    operatorTime
from ods_ower
WHERE id IS NOT NULL
  AND roomId IS NOT NULL
*/
        // 3.插入数据
        tEnv.executeSql(
                "insert into ods_ower_clean\n" +
                        "select\n" +
                        "    id,\n" +
                        "    roomId,\n" +
                        "    owerCode,\n" +
                        "    owerName,\n" +
                        "    owerSex,\n" +
                        "    owerAge,\n" +
                        "    owerIdcard,\n" +
                        "    owerTel,\n" +
                        "    owerHomeplace,\n" +
                        "    owerWorkplace,\n" +
                        "    remark,\n" +
                        "    createTime,\n" +
                        "    operatorTime\n" +
                        "from ods_ower\n" +
                        "WHERE id IS NOT NULL\n" +
                        "  AND roomId IS NOT NULL"
        );

        env.execute("Ower");
    }
}
